package com.navinfo.platform.insurance.service;

import com.google.common.collect.Lists;
import com.navinfo.platform.basicdatastatistics.utils.BigDataUtils;
import com.navinfo.platform.common.service.ICommonService;
import com.navinfo.platform.common.service.IDataHandleService;
import com.navinfo.platform.insurance.dto.DrivingInfo;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.MapPartitionsFunction;
import org.apache.spark.broadcast.Broadcast;
import org.apache.spark.sql.*;
import scala.collection.mutable.WrappedArray;

import java.time.DayOfWeek;
import java.time.LocalDate;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Consumer;

public enum DrivingInfoDataHandleService implements IDataHandleService {
    INS;
    @Override
    public void handle(List<Object> obj, Map<String, String> configMap, SparkSession spark) {
//        LocalDate day = LocalDate.parse(configMap.get(ICommonService.DAY_VALUE), DateTimeFormatter.ofPattern("yyyyMMdd"));
//        Broadcast<LocalDate> dayVar = new JavaSparkContext(spark.sparkContext()).broadcast(day);

        Dataset<DrivingInfo> drivingInfoDS = ((Dataset)obj.get(0)).mapPartitions((MapPartitionsFunction<Row, DrivingInfo>) input -> {
            List<DrivingInfo>  list = new ArrayList<>();
            while(input.hasNext()){
                Row row = input.next();

                DrivingInfo drivingInfo = new DrivingInfo();
                drivingInfo.setTid(row.getAs("tid"));
                drivingInfo.setVin(null);
                drivingInfo.setOperate_days(((Long)row.getAs("days")).shortValue());

                //工作日处理
                WrappedArray.ofRef<String> daysDetail = row.getAs("daysDetail");
                String[] dayArr = daysDetail.array();
                short weekdayCount = 0;
                LocalDate startDate = null;
                LocalDate endDate = null;
                DateTimeFormatter df = DateTimeFormatter.ofPattern("yyyyMMdd");
                for(int i=0; i<dayArr.length; i++){
                    LocalDate tmpDate = LocalDate.parse(dayArr[i], df);
                    if(!DayOfWeek.SATURDAY.equals(tmpDate.getDayOfWeek()) && !DayOfWeek.SUNDAY.equals(tmpDate.getDayOfWeek()) ){
                        weekdayCount++;
                    }
                    if( i==0 ){
                        startDate = tmpDate;
                        endDate = tmpDate;
                    }else{
                        if(startDate.isAfter(tmpDate)){
                            startDate = tmpDate;
                        }
                        if(endDate.isBefore(tmpDate)){
                            endDate = tmpDate;
                        }
                    }
                }

                drivingInfo.setWeekday_days(weekdayCount);
                drivingInfo.setStart_day(df.format(startDate));
                drivingInfo.setEnd_day(df.format(endDate));
                drivingInfo.setMileage(row.getAs("mileage"));
                drivingInfo.setDuration(((Long)row.getAs("duration")).intValue());

                Optional.ofNullable((Long)row.getAs("fatigue_duration")).ifPresent(e -> {
                    drivingInfo.setFatigue_duration(e.intValue());
                });

                Integer[] mileage_details = ((WrappedArray.ofRef<Integer>) row.getAs("mileage_detail")).array();
                drivingInfo.setMileage_detail(mileage_details);

                //计算四百以内，四百至八百公里 疲劳驾驶 行程次数
                Integer[] fatigue_number_details = ((WrappedArray.ofRef<Integer>) row.getAs("fatigue_number_detail")).array();
                int trip400_count = 0,trip800_count=0,fatigue400_count=0,fatigue800_count=0;
                for(int i=0; i<mileage_details.length; i++){
                    int mile = mileage_details[i];
                    int fatigue_number = fatigue_number_details[i];
                    if(mile < 400*1000L){
                        trip400_count++;
                        fatigue400_count += fatigue_number;
                    }else if(mile>=400*1000L && mile < 800*1000L){
                        trip800_count++;
                        fatigue800_count += fatigue_number;
                    }
                }
                drivingInfo.setTrip400_count(trip400_count);
                drivingInfo.setTrip800_count(trip800_count);
                drivingInfo.setFatigue400_count(fatigue400_count);
                drivingInfo.setFatigue800_count(fatigue800_count);

                //超载总次数
                AtomicInteger overload_count =new AtomicInteger(0);
                WrappedArray.ofRef<Row>  wload_type_detail = row.getAs("wload_type_detail");
                Arrays.stream(wload_type_detail.array()).forEach(item -> {
                    Optional.ofNullable((Row) item.getAs("type3")).ifPresent(type3-> {
                        Optional.ofNullable((Integer)type3.getAs("mileage")).ifPresent(e -> {
                            if(null!=e && e>0){
                                overload_count.incrementAndGet();
                            }
                        });
                    });
                });
                drivingInfo.setOverload_count(overload_count.get());

                Optional.ofNullable((Long)row.getAs("sharp_brake_count")).ifPresent(e -> {
                    drivingInfo.setSharp_brake_count(e.intValue());
                });
                Optional.ofNullable((Long)row.getAs("sharp_accelerator_count")).ifPresent(e -> {
                    drivingInfo.setSharp_accelerator_count(e.intValue());
                });
                Optional.ofNullable((Long)row.getAs("ng_coast_count")).ifPresent(e -> {
                    drivingInfo.setNg_coast_count(e.intValue());
                });
                Optional.ofNullable((Long)row.getAs("ig_coast_count")).ifPresent(e -> {
                    drivingInfo.setIg_coast_count(e.intValue());
                });

                //处理中午或晚上的里程时长统计
                WrappedArray.ofRef<Row> vehicle_hour_detail = row.getAs("vehicle_hour_detail");
                AtomicInteger noonDuration = new AtomicInteger();
                AtomicLong noonMileage = new AtomicLong();
                AtomicInteger nightDuration = new AtomicInteger();
                AtomicLong nightMileage = new AtomicLong();
                Arrays.stream(vehicle_hour_detail.array()).forEach(hourRowConsumer(noonDuration, noonMileage, nightDuration, nightMileage));
                drivingInfo.setNoon_duration(noonDuration.get());
                drivingInfo.setNoon_mileage(noonMileage.get());
                drivingInfo.setNight_duration(nightDuration.get());
                drivingInfo.setNight_mileage(nightMileage.get());

                //新增需求
                dealNewColumn(row, drivingInfo);

                list.add(drivingInfo);
            }
            return list.iterator();
        }, Encoders.bean(DrivingInfo.class));

        //保存数据到表中
        String tableName = configMap.get("hive.insurance.info.table");
        String columns = configMap.get("hive.insurance.info.columns");
        String saveHiveDbName = configMap.get("hive.insurance.db");
        List<String> partationKeys = Collections.singletonList("day");
        List<Object> partationValues = Collections.singletonList(configMap.get(ICommonService.DAY_VALUE));
        BigDataUtils.saveDataSetToHiveTable(spark, drivingInfoDS, BigDataUtils.CoalesceType.repartation, 1, saveHiveDbName, tableName, partationKeys, partationValues,  Lists.newArrayList(columns.split(",")));
    }

    /**
     *  新增字段需求
     * @param row
     * @param drivingInfo
     */
    private void dealNewColumn(Row row, DrivingInfo drivingInfo) {
        drivingInfo.setFuel(Optional.ofNullable((Long) row.getAs("fuel")).orElse(0L).intValue());
        drivingInfo.setFout_count(Optional.ofNullable((Long) row.getAs("fout_count")).orElse(0L).intValue());
        drivingInfo.setShift_gears_count(Optional.ofNullable((Long) row.getAs("shift_gears_count")).orElse(0L).intValue());
        drivingInfo.setSharp_turning_count(Optional.ofNullable((Long) row.getAs("sharp_turning_count")).orElse(0L).intValue());
        drivingInfo.setIdle_duration(Optional.ofNullable((Long) row.getAs("idle_duration")).orElse(0L).intValue());
        drivingInfo.setEngine_work_duration(Optional.ofNullable((Long) row.getAs("engine_work_duration")).orElse(0L).intValue());
        drivingInfo.setIg_coast_mileage(row.getAs("ig_coast_mileage"));
        drivingInfo.setIg_coast_duration(Optional.ofNullable((Long) row.getAs("ig_coast_duration")).orElse(0L).intValue());
        drivingInfo.setBrake_mileage(row.getAs("brake_mileage"));
        drivingInfo.setBrake_duration(Optional.ofNullable((Long) row.getAs("brake_duration")).orElse(0L).intValue());
        drivingInfo.setCity_driving_mileage(row.getAs("city_driving_mileage"));
        drivingInfo.setCity_driving_duration(Optional.ofNullable((Long) row.getAs("city_driving_duration")).orElse(0L).intValue());
        drivingInfo.setRural_driving_mileage(row.getAs("rural_driving_mileage"));
        drivingInfo.setRural_driving_duration(Optional.ofNullable((Long) row.getAs("rural_driving_duration")).orElse(0L).intValue());
        drivingInfo.setEco_driving_mileage(row.getAs("eco_driving_mileage"));
        drivingInfo.setEco_driving_duration(Optional.ofNullable((Long) row.getAs("eco_driving_duration")).orElse(0L).intValue());
        drivingInfo.setSharp_brake_mileage(row.getAs("sharp_brake_mileage"));
        drivingInfo.setSharp_brake_duration(Optional.ofNullable((Long) row.getAs("sharp_brake_duration")).orElse(0L).intValue());
        drivingInfo.setSharp_up_speed_mileage(row.getAs("sharp_up_speed_mileage"));
        drivingInfo.setSharp_up_speed_duration(Optional.ofNullable((Long) row.getAs("sharp_up_speed_duration")).orElse(0L).intValue());
        drivingInfo.setSharp_up_speed_count(Optional.ofNullable((Long) row.getAs("sharp_up_speed_count")).orElse(0L).intValue());
        drivingInfo.setSharp_accelerator_mileage(row.getAs("sharp_accelerator_mileage"));
        drivingInfo.setSharp_accelerator_duration(Optional.ofNullable((Long) row.getAs("sharp_accelerator_duration")).orElse(0L).intValue());
        drivingInfo.setVehicle_start_count(Optional.ofNullable((Long) row.getAs("vehicle_start_count")).orElse(0L).intValue());
        drivingInfo.setVehicle_cold_start_count(Optional.ofNullable((Long) row.getAs("vehicle_cold_start_count")).orElse(0L).intValue());
        drivingInfo.setAssist_brake_mileage(row.getAs("assist_brake_mileage"));
        drivingInfo.setAssist_brake_duration(Optional.ofNullable((Long) row.getAs("assist_brake_duration")).orElse(0L).intValue());
        drivingInfo.setSpeed_range01_mileage(row.getAs("speed_range01_mileage"));
        drivingInfo.setSpeed_range01_duration(Optional.ofNullable((Long) row.getAs("speed_range01_duration")).orElse(0L).intValue());
        drivingInfo.setSpeed_range02_mileage(row.getAs("speed_range02_mileage"));
        drivingInfo.setSpeed_range02_duration(Optional.ofNullable((Long) row.getAs("speed_range02_duration")).orElse(0L).intValue());
        drivingInfo.setSpeed_range03_mileage(row.getAs("speed_range03_mileage"));
        drivingInfo.setSpeed_range03_duration(Optional.ofNullable((Long) row.getAs("speed_range03_duration")).orElse(0L).intValue());
        drivingInfo.setSpeed_range04_mileage(row.getAs("speed_range04_mileage"));
        drivingInfo.setSpeed_range04_duration(Optional.ofNullable((Long) row.getAs("speed_range04_duration")).orElse(0L).intValue());
        drivingInfo.setSpeed_range05_mileage(row.getAs("speed_range05_mileage"));
        drivingInfo.setSpeed_range05_duration(Optional.ofNullable((Long) row.getAs("speed_range05_duration")).orElse(0L).intValue());
        drivingInfo.setSpeed_range06_mileage(row.getAs("speed_range06_mileage"));
        drivingInfo.setSpeed_range06_duration(Optional.ofNullable((Long) row.getAs("speed_range06_duration")).orElse(0L).intValue());
        drivingInfo.setSpeed_range07_mileage(row.getAs("speed_range07_mileage"));
        drivingInfo.setSpeed_range07_duration(Optional.ofNullable((Long) row.getAs("speed_range07_duration")).orElse(0L).intValue());
        drivingInfo.setSpeed_range08_mileage(row.getAs("speed_range08_mileage"));
        drivingInfo.setSpeed_range08_duration(Optional.ofNullable((Long) row.getAs("speed_range08_duration")).orElse(0L).intValue());
        drivingInfo.setSpeed_range09_mileage(row.getAs("speed_range09_mileage"));
        drivingInfo.setSpeed_range09_duration(Optional.ofNullable((Long) row.getAs("speed_range09_duration")).orElse(0L).intValue());
        drivingInfo.setSpeed_range10_mileage(row.getAs("speed_range10_mileage"));
        drivingInfo.setSpeed_range10_duration(Optional.ofNullable((Long) row.getAs("speed_range10_duration")).orElse(0L).intValue());
        drivingInfo.setSpeed_range11_mileage(row.getAs("speed_range11_mileage"));
        drivingInfo.setSpeed_range11_duration(Optional.ofNullable((Long) row.getAs("speed_range11_duration")).orElse(0L).intValue());
        drivingInfo.setSpeed_range12_mileage(row.getAs("speed_range12_mileage"));
        drivingInfo.setSpeed_range12_duration(Optional.ofNullable((Long) row.getAs("speed_range12_duration")).orElse(0L).intValue());
        drivingInfo.setGear01_mileage(row.getAs("gear01_mileage"));
        drivingInfo.setGear01_duration(Optional.ofNullable((Long) row.getAs("gear01_duration")).orElse(0L).intValue());
        drivingInfo.setGear02_mileage(row.getAs("gear02_mileage"));
        drivingInfo.setGear02_duration(Optional.ofNullable((Long) row.getAs("gear02_duration")).orElse(0L).intValue());
        drivingInfo.setGear03_mileage(row.getAs("gear03_mileage"));
        drivingInfo.setGear03_duration(Optional.ofNullable((Long) row.getAs("gear03_duration")).orElse(0L).intValue());
        drivingInfo.setGear04_mileage(row.getAs("gear04_mileage"));
        drivingInfo.setGear04_duration(Optional.ofNullable((Long) row.getAs("gear04_duration")).orElse(0L).intValue());
        drivingInfo.setGear05_mileage(row.getAs("gear05_mileage"));
        drivingInfo.setGear05_duration(Optional.ofNullable((Long) row.getAs("gear05_duration")).orElse(0L).intValue());
        drivingInfo.setGear06_mileage(row.getAs("gear06_mileage"));
        drivingInfo.setGear06_duration(Optional.ofNullable((Long) row.getAs("gear06_duration")).orElse(0L).intValue());
        drivingInfo.setGear07_mileage(row.getAs("gear07_mileage"));
        drivingInfo.setGear07_duration(Optional.ofNullable((Long) row.getAs("gear07_duration")).orElse(0L).intValue());
        drivingInfo.setGear08_mileage(row.getAs("gear08_mileage"));
        drivingInfo.setGear08_duration(Optional.ofNullable((Long) row.getAs("gear08_duration")).orElse(0L).intValue());
        drivingInfo.setGear09_mileage(row.getAs("gear09_mileage"));
        drivingInfo.setGear09_duration(Optional.ofNullable((Long) row.getAs("gear09_duration")).orElse(0L).intValue());
        drivingInfo.setGear10_mileage(row.getAs("gear10_mileage"));
        drivingInfo.setGear10_duration(Optional.ofNullable((Long) row.getAs("gear10_duration")).orElse(0L).intValue());
        drivingInfo.setGear11_mileage(row.getAs("gear11_mileage"));
        drivingInfo.setGear11_duration(Optional.ofNullable((Long) row.getAs("gear11_duration")).orElse(0L).intValue());
        drivingInfo.setGear12_mileage(row.getAs("gear12_mileage"));
        drivingInfo.setGear12_duration(Optional.ofNullable((Long) row.getAs("gear12_duration")).orElse(0L).intValue());
        drivingInfo.setGear13_mileage(row.getAs("gear13_mileage"));
        drivingInfo.setGear13_duration(Optional.ofNullable((Long) row.getAs("gear13_duration")).orElse(0L).intValue());
        drivingInfo.setGear14_mileage(row.getAs("gear14_mileage"));
        drivingInfo.setGear14_duration(Optional.ofNullable((Long) row.getAs("gear14_duration")).orElse(0L).intValue());
        drivingInfo.setGear15_mileage(row.getAs("gear15_mileage"));
        drivingInfo.setGear15_duration(Optional.ofNullable((Long) row.getAs("gear15_duration")).orElse(0L).intValue());
        drivingInfo.setGear16_mileage(row.getAs("gear16_mileage"));
        drivingInfo.setGear16_duration(Optional.ofNullable((Long) row.getAs("gear16_duration")).orElse(0L).intValue());
        drivingInfo.setGear_other_mileage(row.getAs("gear_other_mileage"));
        drivingInfo.setGear_other_duration(Optional.ofNullable((Long) row.getAs("gear_other_duration")).orElse(0L).intValue());
        drivingInfo.setRpm_range500_mileage(row.getAs("rpm_range500_mileage"));
        drivingInfo.setRpm_range500_duration(Optional.ofNullable((Long) row.getAs("rpm_range500_duration")).orElse(0L).intValue());
        drivingInfo.setRpm_range800_mileage(row.getAs("rpm_range800_mileage"));
        drivingInfo.setRpm_range800_duration(Optional.ofNullable((Long) row.getAs("rpm_range800_duration")).orElse(0L).intValue());
        drivingInfo.setRpm_range1100_mileage(row.getAs("rpm_range1100_mileage"));
        drivingInfo.setRpm_range1100_duration(Optional.ofNullable((Long) row.getAs("rpm_range1100_duration")).orElse(0L).intValue());
        drivingInfo.setRpm_range1400_mileage(row.getAs("rpm_range1400_mileage"));
        drivingInfo.setRpm_range1400_duration(Optional.ofNullable((Long) row.getAs("rpm_range1400_duration")).orElse(0L).intValue());
        drivingInfo.setRpm_range1700_mileage(row.getAs("rpm_range1700_mileage"));
        drivingInfo.setRpm_range1700_duration(Optional.ofNullable((Long) row.getAs("rpm_range1700_duration")).orElse(0L).intValue());
        drivingInfo.setRpm_range2000_mileage(row.getAs("rpm_range2000_mileage"));
        drivingInfo.setRpm_range2000_duration(Optional.ofNullable((Long) row.getAs("rpm_range2000_duration")).orElse(0L).intValue());
        drivingInfo.setRpm_range2300_mileage(row.getAs("rpm_range2300_mileage"));
        drivingInfo.setRpm_range2300_duration(Optional.ofNullable((Long) row.getAs("rpm_range2300_duration")).orElse(0L).intValue());
        drivingInfo.setRpm_range2600_mileage(row.getAs("rpm_range2600_mileage"));
        drivingInfo.setRpm_range2600_duration(Optional.ofNullable((Long) row.getAs("rpm_range2600_duration")).orElse(0L).intValue());
        drivingInfo.setRpm_rangeGT2600_mileage(row.getAs("rpm_rangeGT2600_mileage"));
        drivingInfo.setRpm_rangeGT2600_duration(Optional.ofNullable((Long) row.getAs("rpm_rangeGT2600_duration")).orElse(0L).intValue());
        drivingInfo.setRpm_sum(Optional.ofNullable((Long) row.getAs("rpm_sum")).orElse(0L).intValue());
        drivingInfo.setRpm_count(Optional.ofNullable((Long) row.getAs("rpm_count")).orElse(0L).intValue());
        drivingInfo.setGas_open0_duration(Optional.ofNullable((Long) row.getAs("gas_open0_duration")).orElse(0L).intValue());
        drivingInfo.setGas_open10_duration(Optional.ofNullable((Long) row.getAs("gas_open10_duration")).orElse(0L).intValue());
        drivingInfo.setGas_open20_duration(Optional.ofNullable((Long) row.getAs("gas_open20_duration")).orElse(0L).intValue());
        drivingInfo.setGas_open30_duration(Optional.ofNullable((Long) row.getAs("gas_open30_duration")).orElse(0L).intValue());
        drivingInfo.setGas_open40_duration(Optional.ofNullable((Long) row.getAs("gas_open40_duration")).orElse(0L).intValue());
        drivingInfo.setGas_open50_duration(Optional.ofNullable((Long) row.getAs("gas_open50_duration")).orElse(0L).intValue());
        drivingInfo.setGas_open60_duration(Optional.ofNullable((Long) row.getAs("gas_open60_duration")).orElse(0L).intValue());
        drivingInfo.setGas_open70_duration(Optional.ofNullable((Long) row.getAs("gas_open70_duration")).orElse(0L).intValue());
        drivingInfo.setGas_open80_duration(Optional.ofNullable((Long) row.getAs("gas_open80_duration")).orElse(0L).intValue());
        drivingInfo.setGas_open90_duration(Optional.ofNullable((Long) row.getAs("gas_open90_duration")).orElse(0L).intValue());
        drivingInfo.setGas_open99_duration(Optional.ofNullable((Long) row.getAs("gas_open99_duration")).orElse(0L).intValue());
        drivingInfo.setGas_open100_duration(Optional.ofNullable((Long) row.getAs("gas_open100_duration")).orElse(0L).intValue());
        drivingInfo.setGas_open_sum(Optional.ofNullable((Long) row.getAs("gas_open_sum")).orElse(0L).intValue());
        drivingInfo.setGas_open_count(Optional.ofNullable((Long) row.getAs("gas_open_count")).orElse(0L).intValue());
        drivingInfo.setTorque0_duration(Optional.ofNullable((Long) row.getAs("torque0_duration")).orElse(0L).intValue());
        drivingInfo.setTorque10_duration(Optional.ofNullable((Long) row.getAs("torque10_duration")).orElse(0L).intValue());
        drivingInfo.setTorque20_duration(Optional.ofNullable((Long) row.getAs("torque20_duration")).orElse(0L).intValue());
        drivingInfo.setTorque30_duration(Optional.ofNullable((Long) row.getAs("torque30_duration")).orElse(0L).intValue());
        drivingInfo.setTorque40_duration(Optional.ofNullable((Long) row.getAs("torque40_duration")).orElse(0L).intValue());
        drivingInfo.setTorque50_duration(Optional.ofNullable((Long) row.getAs("torque50_duration")).orElse(0L).intValue());
        drivingInfo.setTorque60_duration(Optional.ofNullable((Long) row.getAs("torque60_duration")).orElse(0L).intValue());
        drivingInfo.setTorque70_duration(Optional.ofNullable((Long) row.getAs("torque70_duration")).orElse(0L).intValue());
        drivingInfo.setTorque80_duration(Optional.ofNullable((Long) row.getAs("torque80_duration")).orElse(0L).intValue());
        drivingInfo.setTorque90_duration(Optional.ofNullable((Long) row.getAs("torque90_duration")).orElse(0L).intValue());
        drivingInfo.setTorque99_duration(Optional.ofNullable((Long) row.getAs("torque99_duration")).orElse(0L).intValue());
        drivingInfo.setTorque100_duration(Optional.ofNullable((Long) row.getAs("torque100_duration")).orElse(0L).intValue());
        drivingInfo.setTorque_sum(Optional.ofNullable((Long) row.getAs("torque_sum")).orElse(0L).intValue());
        drivingInfo.setTorque_count(Optional.ofNullable((Long) row.getAs("torque_count")).orElse(0L).intValue());
        drivingInfo.setIat50_duration(Optional.ofNullable((Long) row.getAs("iat50_duration")).orElse(0L).intValue());
        drivingInfo.setIat70_duration(Optional.ofNullable((Long) row.getAs("iat70_duration")).orElse(0L).intValue());
        drivingInfo.setIat80_duration(Optional.ofNullable((Long) row.getAs("iat80_duration")).orElse(0L).intValue());
        drivingInfo.setIat85_duration(Optional.ofNullable((Long) row.getAs("iat85_duration")).orElse(0L).intValue());
        drivingInfo.setIat90_duration(Optional.ofNullable((Long) row.getAs("iat90_duration")).orElse(0L).intValue());
        drivingInfo.setIat95_duration(Optional.ofNullable((Long) row.getAs("iat95_duration")).orElse(0L).intValue());
        drivingInfo.setIat100_duration(Optional.ofNullable((Long) row.getAs("iat100_duration")).orElse(0L).intValue());
        drivingInfo.setIat105_duration(Optional.ofNullable((Long) row.getAs("iat105_duration")).orElse(0L).intValue());
        drivingInfo.setIat110_duration(Optional.ofNullable((Long) row.getAs("iat110_duration")).orElse(0L).intValue());
        drivingInfo.setIat115_duration(Optional.ofNullable((Long) row.getAs("iat115_duration")).orElse(0L).intValue());
        drivingInfo.setIat120_duration(Optional.ofNullable((Long) row.getAs("iat120_duration")).orElse(0L).intValue());
        drivingInfo.setIat125_duration(Optional.ofNullable((Long) row.getAs("iat125_duration")).orElse(0L).intValue());
        drivingInfo.setIat130_duration(Optional.ofNullable((Long) row.getAs("iat130_duration")).orElse(0L).intValue());
        drivingInfo.setIat135_duration(Optional.ofNullable((Long) row.getAs("iat135_duration")).orElse(0L).intValue());
        drivingInfo.setIat140_duration(Optional.ofNullable((Long) row.getAs("iat140_duration")).orElse(0L).intValue());
        drivingInfo.setIat145_duration(Optional.ofNullable((Long) row.getAs("iat145_duration")).orElse(0L).intValue());
        drivingInfo.setIat150_duration(Optional.ofNullable((Long) row.getAs("iat150_duration")).orElse(0L).intValue());
        drivingInfo.setIatGT150_duration(Optional.ofNullable((Long) row.getAs("iatGT150_duration")).orElse(0L).intValue());
        drivingInfo.setIat_sum(Optional.ofNullable((Long) row.getAs("iat_sum")).orElse(0L).intValue());
        drivingInfo.setIat_count(Optional.ofNullable((Long) row.getAs("iat_count")).orElse(0L).intValue());
        drivingInfo.setIp50_duration(Optional.ofNullable((Long) row.getAs("ip50_duration")).orElse(0L).intValue());
        drivingInfo.setIp60_duration(Optional.ofNullable((Long) row.getAs("ip60_duration")).orElse(0L).intValue());
        drivingInfo.setIp70_duration(Optional.ofNullable((Long) row.getAs("ip70_duration")).orElse(0L).intValue());
        drivingInfo.setIp80_duration(Optional.ofNullable((Long) row.getAs("ip80_duration")).orElse(0L).intValue());
        drivingInfo.setIp90_duration(Optional.ofNullable((Long) row.getAs("ip90_duration")).orElse(0L).intValue());
        drivingInfo.setIp100_duration(Optional.ofNullable((Long) row.getAs("ip100_duration")).orElse(0L).intValue());
        drivingInfo.setIp110_duration(Optional.ofNullable((Long) row.getAs("ip110_duration")).orElse(0L).intValue());
        drivingInfo.setIp120_duration(Optional.ofNullable((Long) row.getAs("ip120_duration")).orElse(0L).intValue());
        drivingInfo.setIp130_duration(Optional.ofNullable((Long) row.getAs("ip130_duration")).orElse(0L).intValue());
        drivingInfo.setIp140_duration(Optional.ofNullable((Long) row.getAs("ip140_duration")).orElse(0L).intValue());
        drivingInfo.setIp150_duration(Optional.ofNullable((Long) row.getAs("ip150_duration")).orElse(0L).intValue());
        drivingInfo.setIpGT150_duration(Optional.ofNullable((Long) row.getAs("ipGT150_duration")).orElse(0L).intValue());
        drivingInfo.setIp_sum(Optional.ofNullable((Long) row.getAs("ip_sum")).orElse(0L).intValue());
        drivingInfo.setIp_count(Optional.ofNullable((Long) row.getAs("ip_count")).orElse(0L).intValue());
    }

    /**
     * 中午、晚上的里程油耗统计
     * @param noonDuration
     * @param noonMileage
     * @param nightDuration
     * @param nightMileage
     * @return
     */
    private Consumer<Row> hourRowConsumer(AtomicInteger noonDuration, AtomicLong noonMileage, AtomicInteger nightDuration, AtomicLong nightMileage) {
        return item -> {
            //处理中午
            Optional.ofNullable((Row) item.getAs("hour11")).ifPresent(rowConsumer(noonDuration, noonMileage));
            Optional.ofNullable((Row) item.getAs("hour12")).ifPresent(rowConsumer(noonDuration, noonMileage));
            Optional.ofNullable((Row) item.getAs("hour13")).ifPresent(rowConsumer(noonDuration, noonMileage));
            Optional.ofNullable((Row) item.getAs("hour14")).ifPresent(rowConsumer(noonDuration, noonMileage));
            Optional.ofNullable((Row) item.getAs("hour15")).ifPresent(rowConsumer(noonDuration, noonMileage));

            //处理晚上
            Optional.ofNullable((Row) item.getAs("hour20")).ifPresent(rowConsumer(nightDuration, nightMileage));
            Optional.ofNullable((Row) item.getAs("hour21")).ifPresent(rowConsumer(nightDuration, nightMileage));
            Optional.ofNullable((Row) item.getAs("hour22")).ifPresent(rowConsumer(nightDuration, nightMileage));
            Optional.ofNullable((Row) item.getAs("hour23")).ifPresent(rowConsumer(nightDuration, nightMileage));
            Optional.ofNullable((Row) item.getAs("hour0")).ifPresent(rowConsumer(nightDuration, nightMileage));
            Optional.ofNullable((Row) item.getAs("hour1")).ifPresent(rowConsumer(nightDuration, nightMileage));
            Optional.ofNullable((Row) item.getAs("hour2")).ifPresent(rowConsumer(nightDuration, nightMileage));
            Optional.ofNullable((Row) item.getAs("hour3")).ifPresent(rowConsumer(nightDuration, nightMileage));
            Optional.ofNullable((Row) item.getAs("hour4")).ifPresent(rowConsumer(nightDuration, nightMileage));
            Optional.ofNullable((Row) item.getAs("hour5")).ifPresent(rowConsumer(nightDuration, nightMileage));
            Optional.ofNullable((Row) item.getAs("hour6")).ifPresent(rowConsumer(nightDuration, nightMileage));
        };
    }

    /**
     * 里程时长统计
     * @param duration
     * @param mileage
     * @return
     */
    private Consumer<Row> rowConsumer(AtomicInteger duration, AtomicLong mileage) {
        return row -> {
            Optional.ofNullable((Integer) row.getAs("time")).ifPresent(item -> {
                duration.addAndGet(item);
            });
            Optional.ofNullable((Integer) row.getAs("mileage")).ifPresent(item -> {
                mileage.addAndGet(item);
            });
        };
    }

}